package com.cmic.plugin.reader.spdapireader;

import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

public class ApiReader extends Reader {

    public static class Job extends Reader.Job {
        private static final Logger LOG = LoggerFactory
                .getLogger(Job.class);

        private Configuration originalConfig = null;


        @Override
        public void init() {
            this.originalConfig = super.getPluginJobConf();
            LOG.info("job.init:{}",originalConfig);
        }



        @Override
        public void destroy() {

        }

        @Override
        public List<Configuration> split(int adviceNumber) {
            List<Configuration> configurations = new ArrayList<Configuration>();
            for (int i = 0; i < adviceNumber; i++) {
                configurations.add(this.originalConfig.clone());
            }
            return configurations;
        }
    }


    public static class Task extends Reader.Task {

        private static final Logger LOG = LoggerFactory
                .getLogger(ApiReader.Task.class);
        //配置文件
        private Configuration readerSliceConfig;

        @Override
        public void init() {
            this.readerSliceConfig = super.getPluginJobConf();
            LOG.info("task.init:{}",readerSliceConfig);

        }

        @Override
        public void destroy() {

        }

        @Override
        public void startRead(RecordSender recordSender) {
            String result= HttpUtil.get(readerSliceConfig.getString("url"));
            LOG.info("result:{}",result);
            JSONObject jsonObject = JSONUtil.parseObj(result);
            String city = jsonObject.getJSONObject("result").getStr("city");
            String info = jsonObject.getJSONObject("result").getJSONObject("realtime").getStr("info");

            Record record = recordSender.createRecord();
            try {
                record.addColumn(new StringColumn(city));
                record.addColumn(new StringColumn(info));
            } catch (Exception e) {
                throw DataXException.asDataXException(ApiReaderErrorCode.ERROR,"构造一个record失败.", e);
            }

            recordSender.sendToWriter(record);
        }
    }


}
